package dyyx.zb;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;

import dyyx.util.TestDataUtil;

public class ZbQueryAndUpdateTest2 {
	
	public static void main(String[] args) throws Exception {

		List<Map<String,Object>> eventsRaw = TestDataUtil.getTestEventList2();
		
		List<Tuple3<String,String,String>> events =  new ArrayList<>();
		
		for(Map<String,Object> item:eventsRaw) {
			String account = item.get("S_S_ACCTNO").toString();
			String eventTime = item.get("D_T_VB_EVENTOCCURTIME").toString();
			String eventjson = JSON.toJSONString(item);
			Tuple3<String,String,String> event  = new Tuple3<String,String,String>(account,eventTime,eventjson);
			events.add(event);
		}
		
		
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		// 不加这个顺序  结果不对
		env.setParallelism(1);

		
		
		DataSet<Tuple3<String,String,String>> dataset = env.fromCollection(events).setParallelism(1);
		
		
		dataset
		
		.groupBy(0)
		.sortGroup(1, Order.ASCENDING)
		.combineGroup(new MyGroupCombineFunction())
	
		
		.print();
		
		// env.execute("ZbQueryAndUpdateTest");
		
		
		
	}
	

	
	private static class MyGroupCombineFunction implements GroupCombineFunction<Tuple3<String,String,String>,String>{
		
		private static final long serialVersionUID = 1L;

		public void combine(Iterable<Tuple3<String,String,String>> values, Collector<String> out) throws Exception{
			for(Tuple3<String,String,String> item:values) {
				out.collect(item.toString());		
			}	
			out.collect("-------");
		}
	}
	
}
